Events

BindAI uses events to expose important runtime activity without requiring application code to be tightly coupled to the component producing that activity. The event system is built around four core pieces:

Event — the base event object.
EventBus — a synchronous publisher/subscriber mechanism.
ExecutionContext — the execution-scoped context that owns an event bus and execution ID.
EventRecorder and EventTrigger — runtime utilities for recording and reacting to events.

Events are useful for logging, diagnostics, monitoring, automation, and application integrations.

Event Architecture

The basic flow is:

Component
    │
    │ publish(Event)
    ▼
EventBus
    │
    ├── subscribed handlers
    │
    ├── wildcard handlers
    │
    ├── EventRecorder
    │
    └── EventTrigger

A component creates an event and publishes it through an EventBus. Subscribers receive the event synchronously. The event system does not require subscribers to know which component produced the event. They only need to subscribe to the relevant event name.

Event Objects

All BindAI events derive from the Event base class. The base event contains three pieces of metadata:

from bindai_core.events import Event


class MyEvent(Event):
    @property
    def name(self) -> str:
        return "my.event"

An event instance contains:

event.payload
event.id
event.timestamp
event.name

Payload

payload is a dictionary containing event-specific information. It defaults to an empty dictionary. Event-specific event classes may expose additional attributes alongside the payload.

Event ID

Every event receives a unique ID when it is created.

event.id

The ID is generated using uuid4().

Timestamp

Every event receives a UTC timestamp:

event.timestamp

The timestamp is created when the event instance is constructed.

Event name

Every event exposes a string name:

event.name

Event names are used by EventBus subscriptions.

EventBus

EventBus is the central event publisher and subscriber mechanism.

from bindai_core.events import EventBus

bus = EventBus()

The bus supports:

subscribing to an event name
unsubscribing from an event name
publishing events
wildcard subscriptions

Subscribing to Events

Subscribe a callable to an event name:

def handle_event(event):
    print(event.name)


bus.subscribe("my.event", handle_event)

The handler receives the complete event object. A handler therefore has access to:

event.name
event.payload
event.id
event.timestamp

Publishing Events

Publish an event through the bus:

bus.publish(event)

The bus finds handlers registered for the event’s name and invokes them. Handlers are invoked synchronously. For example:

def handle_event(event):
    print("Received:", event.name)

bus.subscribe("my.event", handle_event)

bus.publish(MyEvent())

The handler executes as part of the publish() call.

Wildcard Subscriptions

The event bus supports the special event name:

A wildcard subscriber receives every event published through the bus.

bus.subscribe("*", handle_event)

This is particularly useful for:

recording events
logging
diagnostics
general-purpose observability integrations

For example:

def log_event(event):
    print(event.timestamp, event.name)


bus.subscribe("*", log_event)

The wildcard handler receives events regardless of their specific event name.

Unsubscribing

Remove a previously registered handler with:

bus.unsubscribe("my.event", handle_event)

The handler must be the same callable that was subscribed. Wildcard subscriptions are removed using:

bus.unsubscribe("*", handle_event)

Handler Isolation

Event handlers are isolated from the publisher. The event bus invokes each handler inside a try/except block. If a handler raises an exception:

the exception is logged;
the event bus continues processing the remaining handlers.

For example:

def failing_handler(event):
    raise RuntimeError("handler failed")


def working_handler(event):
    print("Still receives the event")


bus.subscribe("my.event", failing_handler)
bus.subscribe("my.event", working_handler)

A failure in failing_handler does not prevent working_handler from being invoked. This behavior is important for diagnostic and observability handlers because a faulty handler should not normally stop event delivery to other subscribers.

Framework Event Types

BindAI defines a collection of standard event names. The current event types include:

Event	Name
Application started	`application.started`
Application stopped	`application.stopped`
Agent started	`agent.started`
Agent finished	`agent.finished`
Workflow started	`workflow.started`
Workflow finished	`workflow.finished`
Node started	`node.started`
Node completed	`node.completed`
Human task created	`human.task.created`
Human task completed	`human.task.completed`
Model request	`model.requested`
Model response	`model.responded`
Tool executed	`tool.executed`
Memory read	`memory.read`
Memory written	`memory.written`
MCP connected	`mcp.connected`
MCP disconnected	`mcp.disconnected`

The event type definitions provide stable names that can be used by subscribers and integrations.

Agent Events

The core agent event types are:

agent.started
agent.finished

The corresponding event classes are:

from bindai_core.events import AgentStartedEvent, AgentFinishedEvent

For example:

bus.subscribe(
    "agent.started",
    lambda event: print("Agent execution started"),
)

bus.subscribe(
    "agent.finished",
    lambda event: print("Agent execution finished"),
)

These events represent the beginning and end of agent execution.

Model Events

Model interaction exposes two event types:

model.requested
model.responded

The event classes are:

from bindai_core.events import ModelRequestEvent, ModelResponseEvent

A model request event contains the associated model request:

event.request

A model response event contains the associated model response:

event.response

These events can be used to observe model activity without changing the provider implementation. For example:

def on_model_request(event):
    print("Model request:", event.request)


bus.subscribe("model.requested", on_model_request)

Tool Events

Tool execution exposes:

tool.executed

The corresponding event class is:

from bindai_core.events import ToolExecutedEvent

The event identifies the tool that was executed:

event.tool_name

For example:

def on_tool_executed(event):
    print("Tool executed:", event.tool_name)


bus.subscribe("tool.executed", on_tool_executed)

This is useful for application-level logging and diagnostics around tool execution.

ExecutionContext

An ExecutionContext represents the context associated with one execution.

from bindai_core.context import ExecutionContext

context = ExecutionContext()

The context contains an execution ID:

context.execution_id

The ID is generated automatically when the context is created. The context also owns an event bus:

context.events

Therefore execution-scoped code can publish or subscribe through:

context.events

The context additionally contains:

execution state
dependency container
registry
variables
metadata
execution data

A simplified example:

context = ExecutionContext()

print(context.execution_id)

context.events.subscribe(
    "model.responded",
    lambda event: print(event.response),
)

Execution IDs

An ExecutionContext always has an execution_id.

context.execution_id

This ID provides a stable identifier for that execution. Some framework events include the execution ID in their event payload. When an event contains:

event.payload["execution_id"]

the event can be associated with the corresponding execution. It is important not to assume that every event automatically contains an execution_id. Event payloads are event-specific, and only events that include the field can be indexed by execution ID by EventRecorder.

EventRecorder

BindAI provides EventRecorder in the runtime package for lightweight in-memory event recording.

from bindai_runtime.event_recorder import EventRecorder

Create a recorder with an existing event bus:

recorder = EventRecorder(bus)

The recorder automatically subscribes to the bus using the wildcard event:

It therefore records every event published through that bus.

Recording Events

Once attached, events are recorded automatically:

recorder = EventRecorder(bus)

bus.publish(event)

The event can then be retrieved:

events = recorder.all()

The recorder stores events in emission order.

Reading All Recorded Events

Use:

events = recorder.all()

The method returns a copy of the recorded event list. For example:

for event in recorder.all():
    print(event.timestamp, event.name)

The returned list can be inspected without modifying the recorder’s internal storage.

Reading Events for an Execution

If events contain an execution_id in their payload, the recorder indexes them by execution. Use:

events = recorder.for_execution(execution_id)

For example:

execution_events = recorder.for_execution(
    context.execution_id,
)

for event in execution_events:
    print(event.name)

Only events whose payload contains a string execution_id are included in this execution-specific index.

Clearing Recorded Events

Remove all recorded events with:

recorder.clear()

This clears both:

the complete event list
the execution-specific event index

After clearing:

recorder.all()

returns an empty list.

Closing an EventRecorder

When the recorder is no longer needed:

recorder.close()

close() unsubscribes the recorder’s handler from the event bus. The recorder therefore stops receiving newly published events. Closing a recorder does not require destroying the event bus itself.

Recorder Lifecycle

A typical recorder lifecycle is:

from bindai_runtime.event_recorder import EventRecorder

recorder = EventRecorder(bus)

# Events are recorded here.

events = recorder.all()

recorder.close()

If the same recorder needs to be reused after closing, it must be attached again through the supported event-bus subscription mechanism rather than assuming that close() keeps it subscribed.

Thread Safety

EventRecorder protects its internal event collections with a lock. Recording, reading, and clearing events therefore use synchronized access to the recorder’s internal state. The recorder remains intentionally lightweight and stores events in memory. It is not a persistent event database.

EventRecorder Example

A simple execution recording pattern is:

from bindai_core.context import ExecutionContext
from bindai_runtime.event_recorder import EventRecorder

context = ExecutionContext()

recorder = EventRecorder(context.events)

# Components can publish events through context.events.
# ...

all_events = recorder.all()

execution_events = recorder.for_execution(
    context.execution_id,
)

print("All events:", len(all_events))
print("Execution events:", len(execution_events))

recorder.close()

The exact set of execution-indexed events depends on which published events include the execution ID in their payload.

EventTrigger

BindAI also provides an event-driven automation primitive:

from bindai_automation.event_trigger import EventTrigger

An EventTrigger connects an event name to a callable target. Create one with:

trigger = EventTrigger(
    bus=bus,
    event_name="model.responded",
    target=handle_response,
)

The target receives the matching event:

def handle_response(event):
    print("Model response received")

Attaching a Trigger

A trigger is not subscribed to the event bus until it is attached:

trigger.attach()

After attachment, matching events invoke the target. For example:

def handle_response(event):
    print("Response received")


trigger = EventTrigger(
    bus=bus,
    event_name="model.responded",
    target=handle_response,
)

trigger.attach()

The trigger subscribes to the configured event name. Calling attach() more than once does not create duplicate subscriptions.

Detaching a Trigger

Detach a trigger with:

trigger.detach()

After detaching, the trigger no longer receives events from that subscription. Calling detach() when the trigger is already detached is safe.

Enabling and Disabling a Trigger

Triggers are enabled by default. Disable a trigger with:

trigger.disable()

Re-enable it with:

trigger.enable()

These methods return the trigger, so they can be chained:

trigger.disable().enable()

Disabled Triggers

Disabling a trigger does not unsubscribe it from the event bus. Instead, the trigger remains attached but ignores incoming events while disabled. This makes it possible to temporarily pause a trigger:

trigger.disable()

# Matching events are ignored.

trigger.enable()

# Matching events are handled again.

Use detach() when the trigger should no longer be connected to the event bus.

EventTrigger Example

A complete example:

from bindai_automation.event_trigger import EventTrigger
from bindai_core.events import EventBus

bus = EventBus()


def handle_model_response(event):
    print("Model response received:", event.response)


trigger = EventTrigger(
    bus=bus,
    event_name="model.responded",
    target=handle_model_response,
)

trigger.attach()

The trigger now reacts to model.responded events published through bus. To temporarily disable it:

trigger.disable()

To resume:

trigger.enable()

To permanently detach it from the bus:

trigger.detach()

EventBus vs EventTrigger

EventBus and EventTrigger solve different parts of the event system.

EventBus

Use EventBus when application code needs to:

publish events
subscribe handlers
unsubscribe handlers
observe multiple event types
register wildcard handlers

Example:

bus.subscribe("tool.executed", handler)

EventTrigger

Use EventTrigger when an event should activate an automation target and the trigger needs an explicit lifecycle:

trigger.attach()
trigger.disable()
trigger.enable()
trigger.detach()

The trigger is built on top of EventBus.

Event Handling Patterns

Logging

A wildcard subscriber can provide simple event logging:

def log_event(event):
    print(
        event.timestamp,
        event.name,
        event.payload,
    )


bus.subscribe("*", log_event)

Targeted monitoring

Subscribe only to the event of interest:

bus.subscribe(
    "tool.executed",
    on_tool_executed,
)

Execution recording

Attach an EventRecorder:

recorder = EventRecorder(bus)

Automation

Attach an EventTrigger:

trigger = EventTrigger(
    bus=bus,
    event_name="model.responded",
    target=handle_response,
)

trigger.attach()

These patterns can be combined.

Keeping Event Handlers Lightweight

Event handlers execute synchronously during event publication. Handlers should therefore avoid unnecessary blocking work. For example, an event handler that performs a long-running operation directly inside publish() can increase the time spent publishing the event. A lightweight handler can instead collect the information needed for later processing. For example:

def record_tool_usage(event):
    print("Tool:", event.tool_name)


bus.subscribe("tool.executed", record_tool_usage)

The handler should generally perform only the work required for its event-processing responsibility.

Event Handler Errors

Because EventBus isolates handler exceptions, a failing handler does not normally prevent other subscribed handlers from running. The event bus logs the handler exception. Application code should still treat handler failures as real operational errors and monitor the corresponding logs.

Event Names

Use the framework’s existing event names when subscribing to framework events. For example:

bus.subscribe("agent.started", handler)
bus.subscribe("model.requested", handler)
bus.subscribe("model.responded", handler)
bus.subscribe("tool.executed", handler)

When defining application-specific events, use clear names that identify the event being emitted. For example:

document.indexed
customer.created
job.completed

Avoid ambiguous names that make event consumers difficult to understand.

Events and Execution

Events are useful because they separate the component performing an operation from the component observing it. For example, a model execution path can publish a model event while an independent subscriber records or monitors that event. The publishing component does not need to know which subscribers exist. This separation can be represented as:

Agent
  │
  ├── ModelRequestEvent
  │
  ▼
EventBus
  │
  ├── logging handler
  ├── recorder
  └── automation trigger

The same event can therefore support several independent consumers.

Events and Automation

EventTrigger provides the bridge between events and automation. The basic pattern is:

EventBus
   │
   │ matching event
   ▼
EventTrigger
   │
   │ target(event)
   ▼
Application action

For example:

trigger = EventTrigger(
    bus=bus,
    event_name="tool.executed",
    target=handle_tool_execution,
)

trigger.attach()

The target is invoked only when the configured event is published and the trigger is enabled.

Testing Event Handlers

Because EventBus is synchronous, event handling can be tested directly. For example:

from bindai_core.events import EventBus

received = []

def handler(event):
    received.append(event)

bus = EventBus()
bus.subscribe("my.event", handler)

bus.publish(MyEvent())

assert len(received) == 1
assert received[0].name == "my.event"

This allows event-driven behavior to be tested without requiring asynchronous infrastructure.

Testing Event Recording

EventRecorder can be used to inspect events emitted during a test:

from bindai_runtime.event_recorder import EventRecorder

recorder = EventRecorder(bus)

bus.publish(MyEvent())

events = recorder.all()

assert len(events) == 1
assert events[0].name == "my.event"

recorder.close()

Execution-specific recording can be tested when an event contains an execution ID:

events = recorder.for_execution(execution_id)

Testing Event Triggers

Triggers can also be tested synchronously:

received = []


def target(event):
    received.append(event)


trigger = EventTrigger(
    bus=bus,
    event_name="my.event",
    target=target,
)

trigger.attach()

bus.publish(MyEvent())

assert len(received) == 1

Disabled behavior can be tested separately:

trigger.disable()

bus.publish(MyEvent())

assert len(received) == 1

After re-enabling:

trigger.enable()

bus.publish(MyEvent())

assert len(received) == 2

Events vs Hooks

Events and hooks provide different integration mechanisms.

Events

Events are published through an EventBus and can have multiple independent subscribers. Use events when multiple components may need to observe the same activity.

bus.subscribe("model.responded", handler)

Agent lifecycle hooks

Agent hooks are attached directly to an agent and participate in the agent lifecycle. They are appropriate when behavior belongs specifically to the lifecycle of that agent. Events are broader and decouple the producer from its consumers.

Events and Callbacks

Agent callbacks are a separate mechanism from the core EventBus. Use the event system when the behavior should consume a published framework event. Use an agent callback when the behavior is specifically associated with the agent’s callback lifecycle. The two mechanisms should not be treated as interchangeable APIs.

Event Data and Payloads

Do not assume that all events have the same payload structure. The base Event guarantees:

event.payload
event.id
event.timestamp
event.name

Specific event classes may expose additional information. For example, model events expose their associated request or response, while tool events expose the tool name. Consumers should therefore use the documented event-specific attributes for the event type they subscribe to.

Observability Architecture

The event system provides the foundation for runtime observation without requiring every consumer to be embedded inside execution code. A typical architecture is:

                    ┌─────────────────┐
                    │    Agent /      │
                    │   Component     │
                    └────────┬────────┘
                             │
                          Event
                             │
                             ▼
                    ┌─────────────────┐
                    │    EventBus     │
                    └────────┬────────┘
                             │
          ┌──────────────────┼──────────────────┐
          │                  │                  │
          ▼                  ▼                  ▼
      Subscriber         Recorder           Trigger
          │                  │                  │
          ▼                  ▼                  ▼
       Logging          Inspection          Automation

This architecture allows the same event stream to support multiple independent concerns.

Best Practices

Prefer targeted subscriptions when possible:

bus.subscribe("tool.executed", handler)

Use wildcard subscriptions when the application genuinely needs every event.

Keep handlers lightweight

Event delivery is synchronous. Avoid unnecessary blocking work inside handlers.

Use EventRecorder for local inspection

EventRecorder is useful for debugging and tests when events need to be inspected after execution.

Close recorders when finished

Detach recorders that are no longer needed:

recorder.close()

Detach unused triggers

Remove triggers that should no longer listen:

trigger.detach()

Disable triggers temporarily

Use:

trigger.disable()

when the trigger should remain attached but temporarily inactive.

Do not assume execution IDs

Only use execution-specific recording for events that actually include execution_id in their payload.

Keep event consumers decoupled

The purpose of the event system is to allow consumers to observe activity without embedding that consumer logic into the component producing the event.

Summary

BindAI’s event system consists of a small set of composable primitives:

Event provides event metadata and payloads.
EventBus publishes events and manages subscriptions.
wildcard subscriptions allow observation of all events on a bus.
handler exceptions are isolated and logged by the event bus.
ExecutionContext provides an execution ID and an execution-scoped EventBus.
framework events cover application, agent, workflow, node, model, tool, memory, MCP, and human-task event names.
EventRecorder stores events in memory and can index events that contain an execution ID.
EventTrigger connects a specific event to an automation target.
triggers can be attached, detached, enabled, and disabled.
event delivery is synchronous.

Together these APIs provide the foundation for event-driven observation and automation while keeping event producers and consumers loosely coupled.

​Events

​Event Architecture

​Event Objects

​Payload

​Event ID

​Timestamp

​Event name

​EventBus

​Subscribing to Events

​Publishing Events

​Wildcard Subscriptions

​Unsubscribing

​Handler Isolation

​Framework Event Types

​Agent Events

​Model Events

​Tool Events

​ExecutionContext

​Execution IDs

​EventRecorder

​Recording Events

​Reading All Recorded Events

​Reading Events for an Execution

​Clearing Recorded Events

​Closing an EventRecorder

​Recorder Lifecycle

​Thread Safety

​EventRecorder Example

​EventTrigger

​Attaching a Trigger

​Detaching a Trigger

​Enabling and Disabling a Trigger

​Disabled Triggers

​EventTrigger Example

​EventBus vs EventTrigger

​EventBus

​EventTrigger

​Event Handling Patterns

​Logging

​Targeted monitoring

​Execution recording

​Automation

​Keeping Event Handlers Lightweight

​Event Handler Errors

​Event Names

​Events and Execution

​Events and Automation

​Testing Event Handlers

​Testing Event Recording

​Testing Event Triggers

​Events vs Hooks

​Events

​Agent lifecycle hooks

​Events and Callbacks

​Event Data and Payloads

​Observability Architecture

​Best Practices

​Subscribe only to events you need

​Keep handlers lightweight

​Use EventRecorder for local inspection

​Close recorders when finished

​Detach unused triggers

​Disable triggers temporarily

​Do not assume execution IDs

​Keep event consumers decoupled

​Summary

Events

Event Architecture

Event Objects

Payload

Event ID

Timestamp

Event name

EventBus

Subscribing to Events

Publishing Events

Wildcard Subscriptions

Unsubscribing

Handler Isolation

Framework Event Types

Agent Events

Model Events

Tool Events

ExecutionContext

Execution IDs

EventRecorder

Recording Events

Reading All Recorded Events

Reading Events for an Execution

Clearing Recorded Events

Closing an EventRecorder

Recorder Lifecycle

Thread Safety

EventRecorder Example

EventTrigger

Attaching a Trigger

Detaching a Trigger

Enabling and Disabling a Trigger

Disabled Triggers

EventTrigger Example

EventBus vs EventTrigger

EventBus

EventTrigger

Event Handling Patterns

Logging

Targeted monitoring

Execution recording

Automation

Keeping Event Handlers Lightweight

Event Handler Errors

Event Names

Events and Execution

Events and Automation

Testing Event Handlers

Testing Event Recording

Testing Event Triggers

Events vs Hooks

Events

Agent lifecycle hooks

Events and Callbacks

Event Data and Payloads

Observability Architecture

Best Practices

Subscribe only to events you need

Keep handlers lightweight

Use EventRecorder for local inspection

Close recorders when finished

Detach unused triggers

Disable triggers temporarily

Do not assume execution IDs

Keep event consumers decoupled

Summary